/**
 * 
 */
package server.connection.listening;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.SelectorProvider;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.log4j.Logger;

import server.EventDispatcher;
import server.conf.Preferences;
import server.connection.Protocol;
import server.connection.protocol.Event;
import server.exception.ServerException;
import server.log.Logging;

/**
 * Thread that manages a set of connections.
 * Within the thread all client actions are handled: from getting package, processing events to sending messages to other clients.
 * 
 * @author Kamil Nowak    
 *  
 */
public class SelectorThread implements Runnable {

	private Logger logger;
	
	private long threadID;

	private Selector selector;
	
	private final List<SocketChannel> pendingSocketChannels = Collections.synchronizedList(new LinkedList<SocketChannel>());
	private final Object pendingSocketChannlesMutex = new Object();
	
	private ByteBuffer buffer;
	private int bufferSize = Integer.valueOf(Preferences.instance().getPreference("buffer_size"));
	
	private final AtomicInteger socketsNumber = new AtomicInteger(0);
	
	private final CharsetDecoder decoder = (Charset.forName(Preferences.instance().getPreference("encoding"))).newDecoder();
	
	private final ConcurrentHashMap<SocketChannel,Long> socket2conId = new ConcurrentHashMap<SocketChannel,Long>(Integer.valueOf(Preferences.instance().getPreference("connections_per_thread")));
	private final ConcurrentHashMap<Long,SocketChannel> conId2socket = new ConcurrentHashMap<Long,SocketChannel>(Integer.valueOf(Preferences.instance().getPreference("connections_per_thread")));
	
	private final List<SocketChannel> socketsToWrite = new LinkedList<SocketChannel>();
	
	private final HashMap<SocketChannel,List<ByteBuffer>> writeDataQueue = new HashMap<SocketChannel,List<ByteBuffer>>(Integer.valueOf(Preferences.instance().getPreference("connections_per_thread")));  
	private final Object writeDataQueueMutex = new Object();
	
	private final Object stopMutex = new Object();

	private boolean stop;

	private volatile boolean isOperating = true;
	
	/**
	 * Creates and initialize thread object. 
	 * 
	 * @param id thread id
	 * @throws ServerException if initialization fails
	 */
	public SelectorThread(long id, boolean stop) throws ServerException {
		threadID = id;
		logger = Logging.instance().getLogger("selector_thread_" + threadID);
		buffer = ByteBuffer.allocate(bufferSize );
		this.stop = stop;
		init();
	}

	private void init() throws ServerException {
		logger.info("Initialize");
		try {
			selector = SelectorProvider.provider().openSelector();
		} catch (IOException e) {
			logger.error("",e);
			throw new ServerException(e);
		}
		
	}

	/**
	 * Runs a thread; 
	 * 
	 */
	public void run() {
		
		logger.debug("Selector thread starts");
		
		while(true) {
			
			try {
				registerPendingSocketChannels();
				sendPendingData();
				selector.select();
				
				for (Iterator<SelectionKey> i = selector.selectedKeys().iterator(); i.hasNext(); ) {
					SelectionKey key = i.next();
					i.remove();
					
					if (!key.isValid()) { logger.debug("Key is not valid " + key.toString()); continue; };
					if (key.isReadable()) read(key);
					if (key.isWritable()) write(key);
					
				}
				
			} catch (IOException e) {
				logger.debug(e);
			} catch (CancelledKeyException e) {
			} catch (Throwable e) {
				logger.fatal("Unexpected error: ",e);
			}
			
			if (stopThread()) break;
		}
		
		
		logger.debug("Exiting");
		
	}



	private boolean stopThread() {
		
		synchronized (stopMutex) {
			if (stop && socketsNumber.get() == 0 && pendingSocketChannels.size() == 0) { 
				isOperating = false;
				return true;
			}
			return false;
			
		}
		
	}

	private void write(SelectionKey key) throws IOException {
		
		SocketChannel socketChannel = (SocketChannel) key.channel();
		List<ByteBuffer> queue;
		
		synchronized (writeDataQueueMutex) {
			queue = writeDataQueue.get(socketChannel);
		}
		if (queue == null) {
			logger.error("Sending: queue for socketChannel does not exists");
			key.cancel();
			return;
		}
		try {
			synchronized (queue) {
				while(!queue.isEmpty()) {
					ByteBuffer buffer = queue.get(0);
					socketChannel.write(buffer);
					
					if (buffer.remaining() > 0) {
						logger.debug("Socket is full");
						break;
					}
					
					queue.remove(0);
				}
				
				if (queue.isEmpty()) 
					key.interestOps(SelectionKey.OP_READ);
			}
		} catch (IOException e) {
			logger.error(e);
			key.cancel();
			Long connectionId = socket2conId.get(socketChannel);
			EventDispatcher.instance().connectionCleanUp(connectionId);
			closeConnection(socketChannel,connectionId);
		}
		
	}

	private void sendPendingData() {
		synchronized (socketsToWrite) {
			SocketChannel socketChannel;
			for (Iterator<SocketChannel> i = socketsToWrite.iterator(); i.hasNext(); ) {				
				socketChannel = i.next();
				i.remove();
				SelectionKey key = socketChannel.keyFor(selector);
				key.interestOps(SelectionKey.OP_WRITE);
			}
			socketsToWrite.clear();
		}
	}

	private void read(SelectionKey key) throws IOException {
		
		SocketChannel socketChannel = (SocketChannel) key.channel();
		buffer.clear();
		int numRead;

		try {
			numRead = socketChannel.read(buffer);
			
			if (numRead == -1) {
				throw new IOException("No data read from a socket");
			}
			
			buffer.flip();
			String message = decoder.decode( buffer ).toString();
			
			Event event = Protocol.instance().parse(message);
			
			event.setConnectionId(socket2conId.get(socketChannel));
			
			EventDispatcher.instance().addEvent(event);
			
		} catch (IOException e) {
			logger.error(e);
			key.cancel();
			Long connectionId = socket2conId.get(socketChannel);
			EventDispatcher.instance().connectionCleanUp(connectionId);
			closeConnection(socketChannel,connectionId);
			//TODO cleanup after user
		} catch (ServerException e) {
			logger.error(e);
		}
		
	}

	private void connectionCleanUp(SocketChannel socketChannel, Long connectionId) {
		
		socket2conId.remove(socketChannel);
		conId2socket.remove(connectionId);
		
		synchronized (writeDataQueueMutex) {
			writeDataQueue.remove(socketChannel);
		}
		
		synchronized (socketsToWrite) {
			socketsToWrite.remove(socketChannel);
		}
		
		socketsNumber.decrementAndGet();
	}

	private void registerPendingSocketChannels() {
		logger.info("Registers pending socket channel");
		synchronized (pendingSocketChannlesMutex) {
			for (Iterator<SocketChannel> i = pendingSocketChannels.iterator(); i.hasNext(); ) {
				try {
					i.next().register(selector,SelectionKey.OP_READ); 
				} catch (ClosedChannelException e) {
					logger.error("Pending socket channel error ",e);	
				} finally {
					i.remove();
				}
			}
		}
	}

	/**
	 * Return number of connections managed by the thread. 
	 * 
	 * @return number of connections
	 */
	public int getSocketsNumber() {
		return socketsNumber.get();
	}

	/**
	 * Invoked by the MainSelectorThread. 
	 * Adds new connections to be servised by the thread. 
	 * 
	 * @param socketChannelDTO transfer object with information about connection
	 * @throws ServerException if comunication problems occur
	 */
	public boolean addSocketChannel(SocketChannelDTO socketChannelDTO) throws ServerException {
		logger.info("Adding SocketChannel: " + socketChannelDTO.toString());
		try {
			synchronized (stopMutex) {
				
				if (!isOperating) return false;
				
				SocketChannel socketChannel = socketChannelDTO.getSocketChannel();
				socketChannel.configureBlocking(false);
				
				socket2conId.put(socketChannel,socketChannelDTO.getConnectionID());
				conId2socket.put(socketChannelDTO.getConnectionID(),socketChannel);
				
				synchronized (pendingSocketChannlesMutex) {
					pendingSocketChannels.add(socketChannel);
				}
				
				synchronized (writeDataQueueMutex) {
					writeDataQueue.put(socketChannel,new LinkedList<ByteBuffer>());
				}
				
				socketsNumber.incrementAndGet();
			}
			
		} catch (IOException e) {
			logger.error("",e);
			throw new ServerException(e);
		}
		
		return true;
	}

	/**
	 * Wakes the selector. 
	 * Should be invoked after addSocketChannel so that thread could register a new client.
	 */
	public void wakeup() {
		this.selector.wakeup();
		
	}

	/**
	 * Sends data to a client with specified id 
	 * 
	 * @param msg message to send
	 * @param connectionId client's connection id
	 */
	public void send(String msg, Long connectionId) {
		
		List<ByteBuffer> data;
		
		SocketChannel socketChannel = conId2socket.get(connectionId);
		
		if (socketChannel == null) {
			logger.error("SocketChannel for connectionId does not exists " + connectionId);
			return;
		}
		
		synchronized (writeDataQueueMutex) {
			data = writeDataQueue.get(socketChannel);
			if (data == null) {
				logger.error("Queue for SocketChannel does not exists " + connectionId);
				return;
			}
			
			synchronized (data) {
				data.add(ByteBuffer.wrap(msg.getBytes()));
			}
			
			synchronized (socketsToWrite) {
				socketsToWrite.add(socketChannel);
			}
		}
			
	}

	private void closeSocket(SocketChannel socketChannel) throws IOException {
		socketChannel.socket().close();
		socketChannel.close();
	}
	
	/**
	 * Closes connecton and removes all data associated with connection id within the thread 
	 * 
	 * @param connectionId connection to clean
	 * @throws IOException if comunication problems occur
	 */
	public void closeConnection(Long connectionId) throws IOException {
		SocketChannel socketChannel = conId2socket.get(connectionId);
		closeConnection(socketChannel,connectionId);
	}
	
	/**
	 * Closes connecton and removes all data associated with socket within the thread
	 * 
	 * @param socketChannel socket to clean
	 * @throws IOException if comunication problems occur
	 */
	public void closeConnection(SocketChannel socketChannel) throws IOException {
		Long connectionId = socket2conId.get(socketChannel);
		closeConnection(socketChannel,connectionId);
	}
	
	/**
	 * Closes connecton and removes all data associated with connection id and socketChannel within the thread
	 * Neither of data and connection has to be obtained.
	 * 
	 * @param socketChannel socket to clean
	 * @param connectionId connection to clean
	 * @throws IOException if comunication problems occur
	 */
	public void closeConnection(SocketChannel socketChannel, Long connectionId) throws IOException {
		connectionCleanUp(socketChannel,connectionId);
		closeSocket(socketChannel);
	}

	public boolean isOperating() {
		synchronized (stopMutex) {
			return isOperating;
		}
	}
	

	
	
}
